#ifndef __PROTOCOL_CLIENT__
#define __PROTOCOL_CLIENT__


#include <stdlib.h>
#include <inttypes.h>
#include <string.h>
#include <errno.h>
#include <sys/uio.h>
#include "anysocket.h"
//#include "st-hack.h"
#include "ursax.h"
#include "log.h"
#include "protocol.h"

#ifdef __cplusplus
extern "C" {
#endif

extern uint16_t _opsn;

#define URSAX_TIMEOUT (10*1000*1000)		// 10 seconds
typedef int (*send_cb_t)(st_netfd_t out_fd, void* buf, size_t count);

static inline void setup_header(struct ProtocolHeader* header, uint16_t op)
{
	header->magic = PROTOCOL_HEADER_MAGIC;
	header->opsn = _opsn++;
	header->op = op;
}

static inline void init_response(void* _req, void* _resp)
{
	struct ProtocolHeader* req = (struct ProtocolHeader*)_req;
	struct ProtocolHeader* resp = (struct ProtocolHeader*)_resp;
	resp->magic = PROTOCOL_HEADER_MAGIC;
	resp->op = req->op;
	resp->opsn = req->opsn;
}

int parse_ip(char *p, uint32_t* out_ip);

static inline uint32_t get_ip(char* p)
{
	uint32_t ip;
	int ret = parse_ip(p, &ip);
	if (ret != 0)
	{
		fprintf(stderr, "invalid IP: %s\n", p);
		exit(-1);
	}
	return ip;
}

int parse_port(char *p, uint16_t* out_port);

static inline uint16_t get_port(char* p)
{
	uint16_t port;
	int ret = parse_port(p, &port);
	if (ret != 0)
	{
		fprintf(stderr, "invalid port: %s\n", p);
		exit(-1);
	}
	return port;
}

st_netfd_t make_connection(uint32_t ip, uint16_t port);

int log_servers(struct CSEP *servers, int n, const char *input);

int gethostbypeer_str(char* peer, char* hostname);
int gethostbypeer(uint32_t ip, char* hostname);
int getsockname_ex(int s, char* hostname);
bool is_unix_socket(int fd);
void get_peer_addr(st_netfd_t stfd, uint32_t* ip, uint16_t* port);
int wait_socket_ready(int fd, bool read, long utime_out);

// 0 for success, -1 for failed
int parse_ip_port(char* s, uint32_t* ip, uint16_t* port, uint16_t default_port);

void log_socket_error(st_netfd_t stfd, const char* part);

ssize_t st_sendfile(st_netfd_t out_fd, int fd, off_t *offset, size_t count, size_t* sends);
int multy_receive_st_readv(st_netfd_t stfd, struct iovec **iov, int *iov_size, uint32_t size, st_utime_t timeout);
ssize_t st_splice_in(st_netfd_t fd_in, int fd_out, loff_t *off_out, size_t count, int pipes[], unsigned int flags);
int operation_send_recv_with_timeout(st_netfd_t stfd, void* req, uint32_t req_size, void* resp, uint32_t resp_size, uint32_t utimeout);
int operation_send_recv(st_netfd_t stfd, void* req, uint32_t req_size, void* resp, uint32_t resp_size);
int operation_send2_recv(st_netfd_t stfd, void* part1, uint32_t part1_size, void* part2, uint32_t part2_size, void* resp, uint32_t resp_size);
int operation_send2x_recv(st_netfd_t stfd, void* part1, uint32_t part1_size, void* part2, uint32_t part2_size, void* resp, uint32_t resp_size, send_cb_t send);
int operation_send3_recv(st_netfd_t stfd,
						void* part1, uint32_t part1_size,
						void* part2, uint32_t part2_size,
						void* part3, uint32_t part3_size,
						void* resp, uint32_t resp_size);

int operation_send2(st_netfd_t stfd, void* part1, uint32_t part1_size, void* part2, uint32_t part2_size, int timeout);
int operation_send2_recv_with_timeout(st_netfd_t stfd, void* part1, uint32_t part1_size, void* part2, uint32_t part2_size, void* resp, uint32_t resp_size, int timeout);
int operation_send_recv_tmout(st_netfd_t stfd, void* req,
								 uint32_t req_size, void* resp,
								 uint32_t resp_size, uint32_t w_tmout, uint32_t r_tmout);
int operation_sendv_recv(st_netfd_t stfd, void* req, uint32_t req_size, struct iovec *wiov, uint32_t vec_num, uint32_t size, void* resp, uint32_t resp_size);

int st_read_fully_smartly(st_netfd_t stfd, void* buf, uint32_t size, uint32_t timeout);

int st_write_fully_smartly(st_netfd_t stfd, void* buf, uint32_t size, uint32_t timeout);

int reply1(st_netfd_t stfd, void* part1, uint32_t part1_size);
int reply2(st_netfd_t stfd, void* part1, uint32_t part1_size, void* part2, uint32_t part2_size);

int recv_drop_or_log(st_netfd_t stfd, size_t size, const char* logstr);

st_netfd_t ursax_connect(uint32_t ip, uint16_t port, bool do_retry);
st_netfd_t ursax_unix_connect(const char *unix_addr);
int check_unix_file_exist(const char *path);

static inline int log_server_message(st_netfd_t stfd, size_t size)
{
	return recv_drop_or_log(stfd, size, "error message received");
}

// add by zhangzm
ssize_t st_readv_fully(st_netfd_t fd, struct iovec *iov, int iov_size,
			st_utime_t timeout);

static inline int consume_bytes(st_netfd_t stfd, size_t size)
{
	return recv_drop_or_log(stfd, size, NULL);
}

#define REPLY_OK(resp) {								\
		(resp).retcode = 0;								\
		(resp).msg.length = 0;							\
		reply1(stfd, &(resp), sizeof(resp));			\
	}

#define REPLY_OK_RETURN(resp) {							\
		REPLY_OK(resp);									\
		return 0;										\
	}

#define REPLY_ERROR_MESSAGE(resp, code, _msg) {			\
		(resp).retcode = code;							\
		static const char error_message[] = {_msg};			\
		(resp).msg.length = sizeof(error_message)-1;	\
		if((resp).msg.length){							\
			LOG_DEBUG("send to remote: %s", _msg);			\
		}												\
		reply2(stfd, (void*)&(resp), sizeof(resp), 		\
			(void*)error_message, (resp).msg.length);	\
	}

#define REPLY_ERROR_MESSAGE_RETURN(resp, code, _msg) {	\
		REPLY_ERROR_MESSAGE(resp, code, _msg);			\
		return code;									\
	}

// higher 16 bits indicates connection-level errors
// lower 16 bits indicates server returned errors
#define CHECK_CONNECTION_ERROR_LOG_RETURN(ret, req, resp)		\
	if (unlikely(ret < 0))										\
		return MK_CONNECTION_ERROR(ret);						\
	if (unlikely((req)->header.opsn != resp.header.opsn)) 		\
	{															\
		LOG_ERROR("operation serial number doesn't match, request %u, response %u, will reconnect here", (req)->header.opsn, resp.header.opsn); \
		LOG_INFO("request op %u response op %u", (req)->header.op, resp.header.op);\
		LOG_INFO("request magic %u response magic %u", (req)->header.magic, resp.header.magic);\
		return OPSN_ERROR;										\
	}															\
	if (unlikely((req)->header.op != resp.header.op))			\
	{															\
		LOG_ERROR("operation type doesn't match, request %u, response %u, will reconnect here", (req)->header.op, -resp.header.op); \
		return OPSN_ERROR;										\
	}

#define CHECK_ERROR_LOG_RETURN(ret, req, resp, stfd) 	\
	CHECK_CONNECTION_ERROR_LOG_RETURN(ret, req, resp)	\
	if (unlikely(resp.retcode != 0))					\
	{													\
		log_server_message(stfd, resp.msg.length);		\
		return MK_OPERATION_ERROR(resp.retcode);		\
	}
// retcode > 0 means some slave chunkserver down, msg will recive

#ifdef __cplusplus
}
#endif

#endif

